package mammuthus.master

import com.google.inject.{Inject, Singleton}
import net.csdn.common.settings.Settings
import net.csdn.common.logging.Loggers
import akka.actor.{ExtendedActorSystem, ActorSystem}
import org.apache.log4j.{Level, Logger}
import com.typesafe.config.ConfigFactory
import java.util
import scala.concurrent.duration.{Duration, FiniteDuration}

/**
 * 3/27/15 WilliamZhu(allwefantasy@gmail.com)
 */
@Singleton
class Akkautil2 @Inject()(conf: Settings) {
  val logger = Loggers.getLogger(classOf[Akkautil2])

  def createActorSystem(host: String, port: Int): (ActorSystem, Int) = {

    val akkaThreads = conf.getAsInt("qe.akka.threads", 40)
    val akkaBatchSize = conf.getAsInt("qe.akka.batchSize", 15)
    val akkaFrameSize = maxFrameSizeBytes
    val akkaTimeout = conf.getAsInt("qe.akka.timeout", 100)
    val akkaLogLifecycleEvents = conf.getAsBoolean("qe.akka.logLifecycleEvents", false)
    val lifecycleEvents = if (akkaLogLifecycleEvents) "on" else "off"
    if (!akkaLogLifecycleEvents) {
      // As a workaround for Akka issue #3787, we coerce the "EndpointWriter" log to be silent.
      // See: https://www.assembla.com/spaces/akka/tickets/3787#/
      Option(Logger.getLogger("akka.remote.EndpointWriter")).map(l => l.setLevel(Level.FATAL))
    }

    val logAkkaConfig = if (conf.getAsBoolean("qe.akka.logAkkaConfig", false)) "on" else "off"

    val akkaHeartBeatPauses = conf.getAsInt("qe.akka.heartbeat.pauses", 600)
    val akkaFailureDetector = conf.getAsDouble("qe.akka.failure-detector.threshold", 300.0)
    val akkaHeartBeatInterval = conf.getAsInt("qe.akka.heartbeat.interval", 1000)



    val akkaConf = ConfigFactory.parseMap(new util.HashMap[String, String]()).withFallback(
      ConfigFactory.parseString(
        s"""
      |akka.daemonic = on
      |akka.loggers = [""akka.event.slf4j.Slf4jLogger""]
      |akka.stdout-loglevel = "ERROR"
      |akka.jvm-exit-on-fatal-error = off
      |akka.remote.transport-failure-detector.heartbeat-interval = $akkaHeartBeatInterval s
      |akka.remote.transport-failure-detector.acceptable-heartbeat-pause = $akkaHeartBeatPauses s
      |akka.remote.transport-failure-detector.threshold = $akkaFailureDetector
      |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
      |akka.remote.netty.tcp.transport-class = "akka.remote.transport.netty.NettyTransport"
      |akka.remote.netty.tcp.hostname = "$host"
      |akka.remote.netty.tcp.port = $port
      |akka.remote.netty.tcp.tcp-nodelay = on
      |akka.remote.netty.tcp.connection-timeout = $akkaTimeout s
      |akka.remote.netty.tcp.maximum-frame-size = ${akkaFrameSize}B
      |akka.remote.netty.tcp.execution-pool-size = $akkaThreads
      |akka.actor.default-dispatcher.throughput = $akkaBatchSize
      |akka.log-config-on-start = $logAkkaConfig
      |akka.remote.log-remote-lifecycle-events = $lifecycleEvents
      |akka.log-dead-letters = $lifecycleEvents
      |akka.log-dead-letters-during-shutdown = $lifecycleEvents
      |akka.master-dispatcher.type=PinnedDispatcher
      |akka.master-dispatcher.executor="thread-pool-executor"
      |akka.actor.deployment.Worker.dispatcher=master-dispatcher
      |akka.actor.deployment.Master.dispatcher=master-dispatcher
      """.stripMargin))
    val actorSystem = ActorSystem(conf.get("eq.actor-system.name", "mammuthus"), akkaConf)
    val provider = actorSystem.asInstanceOf[ExtendedActorSystem].provider
    val boundPort = provider.getDefaultAddress.port.get
    logger.info(s"boundPort=>${boundPort}")
    (actorSystem, boundPort)
  }

  /** Returns the default Spark timeout to use for Akka ask operations. */
  def askTimeout: FiniteDuration = {
    Duration.create(conf.getAsLong("qe.akka.askTimeout", 30), "seconds")
  }

  /** Returns the default Spark timeout to use for Akka remote actor lookup. */
  def lookupTimeout: FiniteDuration = {
    Duration.create(conf.getAsLong("qe.akka.lookupTimeout", 30), "seconds")
  }

  def maxFrameSizeBytes = {
    conf.getAsInt("qe.akka.frameSize", 10) * 1024 * 1024
  }
}
